package com.huhong.demo.apps

import java.nio.ByteBuffer
import java.util
import java.util.concurrent.{CountDownLatch, ThreadLocalRandom}

import org.apache.zookeeper.Watcher.Event.KeeperState
import org.apache.zookeeper._

import scala.io.StdIn
import scala.util.control.Breaks._
import scala.collection.JavaConversions._

/**
  * Created by huhong on 16/7/22.
  */
object ZookeepApp extends App with Watcher {


  private val LOCK_NODE = "guid-lock-"
  private val count = new CountDownLatch(1)
  val zk = new ZooKeeper("localhost:3000", Int.MaxValue, this)


  count.await()

  println("连接到zookeeper")

  private val rootLockNode: String = "/d_trans"
  if (zk.exists(rootLockNode, false) == null) {
    zk.create(rootLockNode, Array[Byte](), ZooDefs.Ids.OPEN_ACL_UNSAFE,
      CreateMode.PERSISTENT)
  }


  val mutex = new Integer(-1)

  @volatile
  private var currentLock: Int = _

  def acquire() = {


    val b = ByteBuffer.allocate(4)
    b.putInt(ThreadLocalRandom.current().nextInt(10))

    val value = b.array()

    val lockName = zk.create(rootLockNode + "/" + LOCK_NODE, value, ZooDefs.Ids.OPEN_ACL_UNSAFE,
      CreateMode.EPHEMERAL_SEQUENTIAL)

    mutex.synchronized {

      breakable {
        while (true) {
          val acquireLock = new Integer(lockName.substring(lockName.lastIndexOf('-') + 1))
          val childLockNode = zk.getChildren(rootLockNode, true)


          val sortedLock = new util.TreeSet[Int]()
          childLockNode.foreach(nn => {
            sortedLock add (nn.substring(nn.lastIndexOf('-') + 1)).toInt
          })

          currentLock = sortedLock.first()

          println(s"当前${currentLock}")
          //如果当前创建的锁的序号是最小的那么认为这个客户端获得了锁
          if (currentLock >= acquireLock) {
            println("thread_name=" + Thread.currentThread().getName() + "|attend lcok|lock_num=" + currentLock);
            break
          } else {
            //没有获得锁则等待下次事件的发生
            println("thread_name=" + Thread.currentThread().getName() + "|wait lcok|lock_num=" + currentLock);
            mutex.wait();
          }
        }
      }

    }
  }


  def release() = {
    val lockName = String.format("%010d", int2Integer(currentLock))
    zk.delete(rootLockNode + "/" + LOCK_NODE + lockName, -1)
    println("thread_name=" + Thread.currentThread().getName() + "|release lcok|lock_num=" + currentLock)
  }

  def process(event: WatchedEvent): Unit = {

    println(event)
    if (event.getState == KeeperState.SyncConnected) {

      count.countDown()

      mutex.synchronized {
        mutex.notify()
      }

    } else {


      mutex.synchronized {
        mutex.notify()
      }
    }
  }

  acquire()


  StdIn.readLine()


  release()
}
